package com.imooc.structured.streaming
import org.apache.spark.sql.SparkSession

/**
  * http://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#quick-example
  *
  * @description Structured Streaming 处理Socket数据
  * @author yuyon26@126.com
  * @date 2018/10/27 11:23
  */
object StructuredNetworkWordCount {
  def main(args: Array[String]): Unit = {

    val spark = SparkSession
      .builder
      .appName("StructuredNetworkWordCount")
      .master("local[2]")
      .getOrCreate()
    spark.sparkContext.setLogLevel("ERROR")

    import spark.implicits._
    //创建一个DataFrame
    val linesDataFrames = spark.readStream
      .format("socket")
      .option("host", "192.168.31.30")
      .option("port", 9999)
      .load()
      .coalesce(1)

    val words = linesDataFrames.as[String].flatMap(_.split(" "))

    val wordCounts = words.groupBy("value").count()

    val query = wordCounts.writeStream
      .outputMode("complete")
      .format("console")
      .start()

    query.awaitTermination()

  }
}
